﻿#[macro_use] extern crate log;
use either::Either::{Left, Right};
use schemars::JsonSchema;
use futures::{StreamExt, TryStreamExt};
use k8s_openapi::api::core::v1::Pod;
use k8s_openapi::api::core::v1::{Service};
use k8s_openapi::api::apps::v1::{Deployment,DeploymentSpec,DeploymentStatus};
use serde_json::json;
use std::time::Duration;
use tokio::time::sleep;
use serde::{Deserialize, Serialize};
use apiexts::CustomResourceDefinition;
use k8s_openapi::apiextensions_apiserver::pkg::apis::apiextensions::v1beta1 as apiexts;
use kube_runtime::{utils::try_flatten_applied, watcher};
use kube::{ self,
    api::{Api, DeleteParams, ListParams, Patch, PatchParams, PostParams, ResourceExt, WatchEvent},
    core::crd::v1beta1::CustomResourceExt,
    Client, CustomResource,
};
use tracing;
use tracing_subscriber;
use anyhow;



//自定义crd定义
// Own custom resource
#[derive(CustomResource, Deserialize, Serialize, Clone, Debug, JsonSchema)]
#[kube(group = "com.demo", version = "v1", kind = "Appservice", namespaced)]
#[kube(apiextensions = "v1beta1")]
pub struct AppserviceSpec {
    image: String, //镜像
    podNum: i32,//pod数量
	containerPort: i32,//容器端口
    port: i32,//外部访问端口
}




#[tokio::main]
async fn main() -> anyhow::Result<()>{
    std::env::set_var("RUST_LOG", "info,kube=debug");
    tracing_subscriber::fmt::init();
    let client = Client::try_default().await?;
    //let namespace = std::env::var("NAMESPACE").unwrap_or("kube-system".into());

    // Manage pods
	/*
	//测试查询pods列表
    let pods: Api<Pod> = Api::namespaced(client, &namespace);
	
	let lp = ListParams::default()
        //.fields(&format!("labels.k8s-app={}", "metrics-server"))
        .timeout(10);
		
	for p in pods.list(&lp).await? {
        tracing::info!("Found Pod: {}", p.name());
    }
	*/
	
	/*
	
	let namespace = std::env::var("NAMESPACE").unwrap_or("demo".into());
    //测试查询appservice列表
    // Manage appservices
    let appservices: Api<Appservice> = Api::namespaced(client.clone(), &namespace);
	
	let lp = ListParams::default()
        .timeout(10);
		
	for p in appservices.list(&lp).await? {
        tracing::info!("Found Appservice: {}", p.name());
    }
	*/
	
	/*测试监控appservices对象
	let namespace = std::env::var("NAMESPACE").unwrap_or("demo".into());
	let appservices: Api<Appservice> = Api::namespaced(client.clone(), &namespace);
	let lp = ListParams::default()
        .timeout(10);
	let watcher = watcher(appservices, lp);
	try_flatten_applied(watcher)
        .try_for_each(|p| async move {
            tracing::info!("Applied: {}", p.name());
            Ok(())
        })
        .await?;
	*/
	
	//监控appservice新增和更新
	let namespace = std::env::var("NAMESPACE").unwrap_or("demo".into());
	let appservices: Api<Appservice> = Api::namespaced(client.clone(), &namespace);
	let lp = ListParams::default()
        .timeout(10);
	
	 loop {
	   let mut stream = appservices.watch(&lp, "").await?.boxed();
	   while let Some(status) = stream.try_next().await? {
	     tracing::info!("status={:?}", status);
         match status {
            WatchEvent::Added(o) => {
                tracing::info!("Added {}", o.name());
				//添加service
				match CreateOrUpdateService(&client ,&o).await{
				     Ok(()) => { },
                     Err(e) =>tracing::error!("{}", e) , 
				}
				//添加deployment
				match CreateOrUpdateDeployment(&client ,&o).await{
				     Ok(()) => { },
                     Err(e) =>tracing::error!("{}", e) , 
				}
            },
            WatchEvent::Modified(o) => {
                tracing::info!("Ready to attach to {}", o.name());
				
				//删除service
				match deleteService(&client, &o).await{
				     Ok(()) => { },
                     Err(e) =>tracing::error!("{}", e)  ,
				}
				//添加service
				match CreateOrUpdateService(&client ,&o).await{
				     Ok(()) => { },
                     Err(e) =>tracing::error!("{}", e)  ,
				}
				//添加deployment
				match CreateOrUpdateDeployment(&client ,&o).await{
				     Ok(()) => { },
                     Err(e) =>tracing::error!("{}", e)  ,
				}
            },
			WatchEvent::Deleted(o) => {
			
			    tracing::info!("Deleted {}", o.name());
				//删除service
				match deleteService(&client, &o).await{
				     Ok(()) => { },
                     Err(e) =>tracing::error!("{}", e)  ,
				}
				
				//删除deployment
				match deleteDeployment(&client ,&o).await{
				     Ok(()) => { },
                     Err(e) =>tracing::error!("{}", e)  ,
				}
			},
            WatchEvent::Error(err) => tracing::error!("{}", err),
            _ => {},
         }
		}
    }
	
	
	Ok(())
}

//添加service对象
async fn CreateOrUpdateService(client : &Client,o : &Appservice) -> anyhow::Result<()>{
   tracing::info!("o.spec.port={}",&o.spec.port);
   tracing::info!("o.spec.containerPort={}",&o.spec.containerPort);
   tracing::info!("o.spec.image={}",&o.spec.image);
   
   // Create Service
    tracing::info!("Creating Service instance {}",o.name());
	// Manage services
    let services: Api<Service> = Api::namespaced(client.clone(), &o.namespace().unwrap());
	
	let service_name =format!("{}-service",&o.name());
	//已存在不用新建
	let lp = ListParams::default().fields(&format!("metadata.name={}", &service_name)); // only want results for our pod
    let mut exists = false;
	let mut changed =false ;
	let mut resourceVersion :String =String::from("");
	for p in services.list(&lp).await? {
       exists =true;
	   
	   if p.spec.as_ref().unwrap().ports[0].node_port.unwrap() !=o.spec.port {
	      changed =true;
		  resourceVersion =p.resource_version().unwrap();
	   }
	   
	   break;
    }
	
	
	 let pp = PostParams::default();
	
	if !exists {
	  
	
	
	//创建service
	let service: Service = serde_json::from_value(json!({
        "apiVersion": "v1",
        "kind": "Service",
        "metadata": { 
		
		  "name": service_name.clone() ,
		  "namespace": o.namespace().unwrap().clone()
		
		},
        "spec": {
		    "selector":{
			   "app":  o.name()
			},
			"type": "NodePort",
            "ports": [{
              "protocol": "TCP",
              "port": o.spec.port.clone(),
			  "nodePort": o.spec.port.clone(),
			  "targetPort": o.spec.containerPort.clone(),
            }],
        }
    }))?;
	
	

	 
      match services.create(&pp, &service).await {
        Ok(o) => {
            let name = o.name();
            tracing::info!("Created {}", &name);
            // wait for it..
            std::thread::sleep(std::time::Duration::from_millis(5_000));
        },
        Err(kube::Error::Api(ae)) => tracing::error!("{}", ae),
        Err(e) =>tracing::error!("{}", e)                ,       
      }
	 }  else {
	
	 if(changed){
	 
	 
	 //创建service
	let service: Service = serde_json::from_value(json!({
        "apiVersion": "v1",
        "kind": "Service",
        "metadata": { 
		
		  "name":service_name.clone(),
		  "namespace": o.namespace().unwrap().clone(),
		  "resourceVersion":resourceVersion.clone()
		
		},
        "spec": {
		    "selector":{
			   "app":  o.name()
			},
			"type": "NodePort",
            "ports": [{
              "protocol": "TCP",
              "port": o.spec.port.clone(),
			  "nodePort": o.spec.port.clone(),
			  "targetPort": o.spec.containerPort.clone(),
            }],
        }
    }))?;
	
	 
	 
	   let name = o.name();
	   match services.replace(&service_name,&pp, &service).await {
        Ok(o) => {
            
            tracing::info!("Update {}", &name);
            // wait for it..
            std::thread::sleep(std::time::Duration::from_millis(5_000));
        },
        Err(kube::Error::Api(ae)) => tracing::error!("{}", ae),
        Err(e) =>tracing::error!("{}", e)  ,                     
      }
	 }
	
	
	}
		Ok(())
	
}
//删除service对象
async fn deleteService(client : &Client,o :&Appservice) -> anyhow::Result<()>{
   tracing::info!("o.spec.port={}",&o.spec.port);
   tracing::info!("o.spec.containerPort={}",&o.spec.containerPort);
   tracing::info!("o.spec.image={}",&o.spec.image);
   
    // Delete Service
    tracing::info!("Deleteing Service instance {}",o.name());
	// Manage services
    let services: Api<Service> = Api::namespaced(client.clone(), &o.namespace().unwrap());
   let service_name =format!("{}-service",&o.name());
    // Delete it
    let dp = DeleteParams::default();
    match services.delete(&service_name, &dp).await {
        Ok(o) => {
            
           o.map_left(|servicedel| {
       
             tracing::info!("Deleting service started: {:?}", servicedel);
           });
            // wait for it..
            std::thread::sleep(std::time::Duration::from_millis(5_000));
        },
        Err(kube::Error::Api(ae)) => tracing::error!("{}", ae),
        Err(e) =>tracing::error!("{}", e) ,             
      }
	
		Ok(())
 }
 
 
 //删除deployment对象
async fn deleteDeployment(client : &Client,o :&Appservice) -> anyhow::Result<()>{
   tracing::info!("o.spec.port={}",&o.spec.port);
   tracing::info!("o.spec.containerPort={}",&o.spec.containerPort);
   tracing::info!("o.spec.image={}",&o.spec.image);
   
    // Delete Deployment
    tracing::info!("Deleteing Deployment instance {}",o.name());
	// Manage deployments
    let deployments: Api<Deployment> = Api::namespaced(client.clone(), &o.namespace().unwrap());
  
    // Delete it
    let dp = DeleteParams::default();
    match  deployments.delete(&o.name(), &dp).await {
        Ok(o) => {
            
           o.map_left(|deploymentdel| {
       
              tracing::info!("Deleting deployment started: {:?}", deploymentdel);
           });
            // wait for it..
            std::thread::sleep(std::time::Duration::from_millis(5_000));
        },
        Err(kube::Error::Api(ae)) => tracing::error!("{}", ae),
        Err(e) =>tracing::error!("{}", e),        
      }
	  
		Ok(())
 }





//添加deployment对象
async fn CreateOrUpdateDeployment(client : &Client,o : &Appservice) -> anyhow::Result<()>{
   tracing::info!("o.spec.port={}",&o.spec.port);
   tracing::info!("o.spec.containerPort={}",&o.spec.containerPort);
   tracing::info!("o.spec.image={}",&o.spec.image);
   
   // Create Deployment
    tracing::info!("Creating Deployment instance {}",o.name());
	// Manage Deployments
    let deployments: Api<Deployment> = Api::namespaced(client.clone(), &o.namespace().unwrap());
	
	//已存在不用新建
	let lp = ListParams::default().fields(&format!("metadata.name={}", o.name())); // only want results for our pod
    let mut exists = false;
	let mut changed =false ;
	for p in deployments.list(&lp).await? {
       exists =true;
	   if p.spec.as_ref().unwrap().replicas.unwrap() != o.spec.podNum || p.spec.as_ref().unwrap().template.spec.as_ref().unwrap().containers[0].ports[0].container_port != o.spec.containerPort  || p.spec.as_ref().unwrap().template.spec.as_ref().unwrap().containers[0].image.as_ref().unwrap().clone() != o.spec.image {
	      changed =true;
	   }
	   break;
    }
	
	//创建deployment
	let deployment: Deployment = serde_json::from_value(json!({
        "apiVersion": "apps/v1",
        "kind": "Deployment",
        "metadata": { 
		
		    "name":o.name() ,
			"namespace": o.namespace().unwrap().clone(),
			"labels" :{
			    "app.kubernetes.io/part-of": o.name(),
				"app.kubernetes.io/managed-by":"appserviceoperator"
			
			},
			"ownerReferences":[{
			    "apiVersion": "com.demo/v1",
				"kind" : "Appservice",
				
				"name" :  o.name(),
				"uid" :o.uid()
			
			}]
		
		},
        "spec": {
		    "selector":{
			   "matchLabels":  {
			     "app": o.name()
			   }
			},
			"replicas": o.spec.podNum.clone(),
            "template": {
			   "metadata": {
			       "labels": {
				      "app":o.name()
				   }
			   },
			   "spec":{
			       "containers":[
				       {
					        "name":  o.name(),
							"image" :  o. spec.image.clone(),
							"ports": [
							  {"containerPort":o. spec.containerPort.clone() }
							]
					   }
				   ]
			   }
			},
        }
    }))?;

	let pp = PostParams::default();
	
	
	
	if !exists {
	  
	
       match deployments.create(&pp, &deployment).await {
        Ok(o) => {
            let name = o.name();
            tracing::info!("Created {}", &name);
            // wait for it..
            std::thread::sleep(std::time::Duration::from_millis(5_000));
        },
        Err(kube::Error::Api(ae)) => error!("{}", ae),
        Err(e) =>error!("{}", e),                       
       }
	} else {
	  if(changed){
	   let name = o.name();
	   match deployments.replace(&name,&pp, &deployment).await {
        Ok(o) => {
            
            tracing::info!("Created {}", &name);
            // wait for it..
            std::thread::sleep(std::time::Duration::from_millis(5_000));
        },
        Err(kube::Error::Api(ae)) => error!("{}", ae),
        Err(e) =>error!("{}", e),                       
       }
	  }
	}
		Ok(())
	
}